/**
 * @file test_latency.cpp
 * @brief 消息延迟性能测试
 *
 * 测试目标：
 * - P50延迟 < 2ms
 * - P99延迟 < 10ms
 * - P999延迟 < 50ms
 */

#include <gtest/gtest.h>
#include <spdlog/spdlog.h>
#include <chrono>
#include <vector>
#include <algorithm>
#include <atomic>
#include <thread>
#include <iostream>
#include <iomanip>
#include "MessageProducer.h"
#include "MessageConsumer.h"
#include "MessageDispatcher.h"
#include "../test_helpers.h"

using namespace std::chrono;
using namespace MessageFramework;
using namespace MessageFramework::Test;

/**
 * @brief 延迟测量Handler
 */
class LatencyHandler : public IMessageHandler {
public:
    struct LatencyRecord {
        int64_t sendTime;      // 发送时间戳（微秒）
        int64_t receiveTime;   // 接收时间戳（微秒）
        int64_t latency;       // 延迟（微秒）
    };

    int handleMessage(const Message& message) override {
        auto receiveTime = duration_cast<microseconds>(
            system_clock::now().time_since_epoch()
        ).count();

        // 从消息中提取发送时间
        try {
            // 发送时间存储在body中
            int64_t sendTime = std::stoll(message.body);
            int64_t latency = receiveTime - sendTime;

            LatencyRecord record{sendTime, receiveTime, latency};

            std::lock_guard<std::mutex> lock(mutex_);
            records_.push_back(record);
            count_++;
        } catch (const std::exception& e) {
            return -1;  // Error
        }

        return 0;  // Success
    }

    std::string getMessageType() const override {
        return "LatencyTest";
    }

    std::string getName() const override {
        return "LatencyHandler";
    }

    int getCount() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return count_;
    }

    std::vector<LatencyRecord> getRecords() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return records_;
    }

    void reset() {
        std::lock_guard<std::mutex> lock(mutex_);
        records_.clear();
        count_ = 0;
    }

private:
    mutable std::mutex mutex_;
    std::vector<LatencyRecord> records_;
    std::atomic<int> count_{0};
};

/**
 * @brief 延迟统计结果
 */
struct LatencyStats {
    int64_t min;        // 最小延迟（微秒）
    int64_t max;        // 最大延迟（微秒）
    int64_t mean;       // 平均延迟（微秒）
    int64_t p50;        // P50延迟（微秒）
    int64_t p90;        // P90延迟（微秒）
    int64_t p99;        // P99延迟（微秒）
    int64_t p999;       // P999延迟（微秒）
    int totalMessages;  // 消息总数

    void print() const {
        std::cout << "\n========== 延迟统计 ==========\n";
        std::cout << "消息总数: " << totalMessages << "\n";
        std::cout << "最小延迟: " << min << " μs (" << min/1000.0 << " ms)\n";
        std::cout << "最大延迟: " << max << " μs (" << max/1000.0 << " ms)\n";
        std::cout << "平均延迟: " << mean << " μs (" << mean/1000.0 << " ms)\n";
        std::cout << "P50 延迟: " << p50 << " μs (" << p50/1000.0 << " ms)\n";
        std::cout << "P90 延迟: " << p90 << " μs (" << p90/1000.0 << " ms)\n";
        std::cout << "P99 延迟: " << p99 << " μs (" << p99/1000.0 << " ms)\n";
        std::cout << "P999延迟: " << p999 << " μs (" << p999/1000.0 << " ms)\n";
        std::cout << "==============================\n\n";
    }
};

/**
 * @brief 计算延迟统计
 */
LatencyStats calculateLatencyStats(const std::vector<LatencyHandler::LatencyRecord>& records) {
    if (records.empty()) {
        return {0, 0, 0, 0, 0, 0, 0, 0};
    }

    // 提取所有延迟值并排序
    std::vector<int64_t> latencies;
    latencies.reserve(records.size());
    for (const auto& record : records) {
        latencies.push_back(record.latency);
    }
    std::sort(latencies.begin(), latencies.end());

    // 计算统计值
    LatencyStats stats;
    stats.totalMessages = latencies.size();
    stats.min = latencies.front();
    stats.max = latencies.back();

    // 计算平均值
    int64_t sum = 0;
    for (int64_t lat : latencies) {
        sum += lat;
    }
    stats.mean = sum / latencies.size();

    // 计算百分位数
    auto percentile = [&latencies](double p) -> int64_t {
        size_t index = static_cast<size_t>(latencies.size() * p);
        if (index >= latencies.size()) index = latencies.size() - 1;
        return latencies[index];
    };

    stats.p50 = percentile(0.50);
    stats.p90 = percentile(0.90);
    stats.p99 = percentile(0.99);
    stats.p999 = percentile(0.999);

    return stats;
}

/**
 * @brief 延迟性能测试
 */
class LatencyPerformanceTest : public ::testing::Test {
protected:
    void SetUp() override {
        TestHelpers::cleanupRedis("test_latency_*");
        pool_ = TestHelpers::getTestRedisPool();
        streamName_ = "test_latency_stream_" + TestHelpers::randomString(8);

        // 配置Producer
        MessageProducer::Config producerConfig;
        producerConfig.publisherId = "LatencyProducer";
        producerConfig.autoSign = false;     // 禁用签名以减少延迟开销
        producerConfig.autoEncrypt = false;  // 禁用加密以测试纯通信延迟
        producer_ = std::make_unique<MessageProducer>(pool_, producerConfig);

        // 配置Consumer
        handler_ = std::make_shared<LatencyHandler>();

        MessageConsumer::Config consumerConfig;
        consumerConfig.streamName = streamName_;
        consumerConfig.consumerGroup = "LatencyGroup";
        consumerConfig.consumerId = "LatencyConsumer";
        consumerConfig.blockMs = 10;  // 短阻塞时间以减少延迟
        consumerConfig.batchSize = 1; // 每次只读1条消息，立即处理
        consumer_ = std::make_unique<MessageConsumer>(pool_, consumerConfig);

        // 配置Dispatcher
        MessageDispatcher::Config dispatcherConfig;
        dispatcherConfig.verifySignature = false;
        dispatcherConfig.autoDecrypt = false;
        dispatcher_ = std::make_unique<MessageDispatcher>(dispatcherConfig);
        dispatcher_->registerHandler("LatencyTest", handler_);

        // 设置Consumer回调
        consumer_->setMessageCallback([this](const Message& msg) {
            auto result = dispatcher_->dispatch(const_cast<Message&>(msg));
            return result.code;
        });
    }

    void TearDown() override {
        if (consumer_) consumer_->stop();
        TestHelpers::cleanupRedis("test_latency_*");
    }

    RedisPool* pool_;
    std::string streamName_;
    std::unique_ptr<MessageProducer> producer_;
    std::unique_ptr<MessageConsumer> consumer_;
    std::unique_ptr<MessageDispatcher> dispatcher_;
    std::shared_ptr<LatencyHandler> handler_;
};

/**
 * @brief 测试基本延迟（少量消息）
 */
TEST_F(LatencyPerformanceTest, BasicLatency) {
    std::cout << "\n========== 基本延迟测试 (100条消息) ==========\n";

    const int messageCount = 100;
    consumer_->start();
    std::this_thread::sleep_for(milliseconds(100));

    // 发送消息
    for (int i = 0; i < messageCount; i++) {
        auto sendTime = duration_cast<microseconds>(
            system_clock::now().time_since_epoch()
        ).count();

        producer_->send(streamName_, "LatencyTest", std::to_string(sendTime));
    }

    // 等待所有消息处理完成
    bool allReceived = TestHelpers::waitFor([this, messageCount]() {
        return handler_->getCount() >= messageCount;
    }, 10000);

    ASSERT_TRUE(allReceived) << "未能在10秒内接收所有消息";

    // 计算统计
    auto records = handler_->getRecords();
    auto stats = calculateLatencyStats(records);
    stats.print();

    // 验证延迟目标
    EXPECT_LT(stats.p50, 2000) << "P50延迟应 < 2ms";
    EXPECT_LT(stats.p99, 10000) << "P99延迟应 < 10ms";
}

/**
 * @brief 测试中等负载延迟
 */
TEST_F(LatencyPerformanceTest, MediumLoadLatency) {
    std::cout << "\n========== 中等负载延迟测试 (1000条消息) ==========\n";

    const int messageCount = 1000;
    consumer_->start();
    std::this_thread::sleep_for(milliseconds(100));

    // 发送消息
    for (int i = 0; i < messageCount; i++) {
        auto sendTime = duration_cast<microseconds>(
            system_clock::now().time_since_epoch()
        ).count();

        producer_->send(streamName_, "LatencyTest", std::to_string(sendTime));
    }

    // 等待所有消息处理完成
    bool allReceived = TestHelpers::waitFor([this, messageCount]() {
        return handler_->getCount() >= messageCount;
    }, 30000);

    ASSERT_TRUE(allReceived) << "未能在30秒内接收所有消息";

    // 计算统计
    auto records = handler_->getRecords();
    auto stats = calculateLatencyStats(records);
    stats.print();

    // 验证延迟目标
    EXPECT_LT(stats.p50, 2000) << "P50延迟应 < 2ms";
    EXPECT_LT(stats.p99, 10000) << "P99延迟应 < 10ms";
    EXPECT_LT(stats.p999, 50000) << "P999延迟应 < 50ms";
}

/**
 * @brief 测试高负载延迟
 */
TEST_F(LatencyPerformanceTest, HighLoadLatency) {
    std::cout << "\n========== 高负载延迟测试 (5000条消息) ==========\n";

    const int messageCount = 5000;
    consumer_->start();
    std::this_thread::sleep_for(milliseconds(100));

    // 发送消息
    auto startTime = high_resolution_clock::now();
    for (int i = 0; i < messageCount; i++) {
        auto sendTime = duration_cast<microseconds>(
            system_clock::now().time_since_epoch()
        ).count();

        producer_->send(streamName_, "LatencyTest", std::to_string(sendTime));
    }
    auto endTime = high_resolution_clock::now();
    auto sendDuration = duration_cast<milliseconds>(endTime - startTime).count();

    std::cout << "发送耗时: " << sendDuration << " ms\n";
    std::cout << "发送速率: " << (messageCount * 1000 / sendDuration) << " msg/s\n";

    // 等待所有消息处理完成
    bool allReceived = TestHelpers::waitFor([this, messageCount]() {
        return handler_->getCount() >= messageCount;
    }, 60000);

    ASSERT_TRUE(allReceived) << "未能在60秒内接收所有消息";

    // 计算统计
    auto records = handler_->getRecords();
    auto stats = calculateLatencyStats(records);
    stats.print();

    // 验证延迟目标
    EXPECT_LT(stats.p50, 2000) << "P50延迟应 < 2ms";
    EXPECT_LT(stats.p99, 10000) << "P99延迟应 < 10ms";
    EXPECT_LT(stats.p999, 50000) << "P999延迟应 < 50ms";
}

/**
 * @brief 测试加密消息延迟
 */
TEST_F(LatencyPerformanceTest, EncryptedLatency) {
    std::cout << "\n========== 加密消息延迟测试 (1000条消息) ==========\n";

    // 重新配置启用加密
    MessageProducer::Config producerConfig;
    producerConfig.publisherId = "LatencyProducerEnc";
    producerConfig.autoSign = false;
    producerConfig.autoEncrypt = true;
    producerConfig.encryptionKey = "test_encryption_key_32_bytes_long!";
    producer_ = std::make_unique<MessageProducer>(pool_, producerConfig);

    MessageDispatcher::Config dispatcherConfig;
    dispatcherConfig.verifySignature = false;
    dispatcherConfig.autoDecrypt = true;
    dispatcherConfig.encryptionKey = "test_encryption_key_32_bytes_long!";
    dispatcher_ = std::make_unique<MessageDispatcher>(dispatcherConfig);
    dispatcher_->registerHandler("LatencyTest", handler_);

    const int messageCount = 1000;
    consumer_->start();
    std::this_thread::sleep_for(milliseconds(100));

    // 发送消息
    for (int i = 0; i < messageCount; i++) {
        auto sendTime = duration_cast<microseconds>(
            system_clock::now().time_since_epoch()
        ).count();

        producer_->send(streamName_, "LatencyTest", std::to_string(sendTime));
    }

    // 等待所有消息处理完成
    bool allReceived = TestHelpers::waitFor([this, messageCount]() {
        return handler_->getCount() >= messageCount;
    }, 30000);

    ASSERT_TRUE(allReceived) << "未能在30秒内接收所有消息";

    // 计算统计
    auto records = handler_->getRecords();
    auto stats = calculateLatencyStats(records);
    stats.print();

    // 加密会增加延迟，但应仍在合理范围内
    EXPECT_LT(stats.p50, 5000) << "加密消息P50延迟应 < 5ms";
    EXPECT_LT(stats.p99, 20000) << "加密消息P99延迟应 < 20ms";
}

int main(int argc, char** argv) {
    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}
